/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.client.consumer;

import java.util.Iterator;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.log.ClientLogger;
import com.alibaba.rocketmq.common.ThreadFactoryImpl;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;

/**
 * Schedule service for pull consumer
 *
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2014-2-26
 */
public class MQPullConsumerScheduleService {
	private final Logger log = ClientLogger.getLog();
	private DefaultMQPullConsumer defaultMQPullConsumer;
	private int pullThreadNums = 20;
	private ConcurrentHashMap<String /* topic */, PullTaskCallback> callbackTable = new ConcurrentHashMap<String, PullTaskCallback>();
	private ScheduledThreadPoolExecutor scheduledThreadPoolExecutor;
	private final MessageQueueListener messageQueueListener = new MessageQueueListenerImpl();

	// 表示正在拉取消息的任务列表
	private final ConcurrentHashMap<MessageQueue, PullTaskImpl> taskTable = new ConcurrentHashMap<MessageQueue, PullTaskImpl>();

	class MessageQueueListenerImpl implements MessageQueueListener {
		@Override
		public void messageQueueChanged(String topic, Set<MessageQueue> mqAll, Set<MessageQueue> mqDivided) {
			MessageModel messageModel = MQPullConsumerScheduleService.this.defaultMQPullConsumer.getMessageModel();
			switch (messageModel) {
				case BROADCASTING :
					// 广播模式 mqNewSet为该topic下面的所有MessageQueue队列
					MQPullConsumerScheduleService.this.putTask(topic, mqAll);
					break;
				case CLUSTERING :
					// 集群模式 mqNewSet为给该topic分配的MessageQueue队列
					MQPullConsumerScheduleService.this.putTask(topic, mqDivided);
					break;
				default :
					break;
			}
		}
	}

	class PullTaskImpl implements Runnable {
		private final MessageQueue messageQueue;
		private volatile boolean cancelled = false;

		public PullTaskImpl(final MessageQueue messageQueue) {
			this.messageQueue = messageQueue;
		}

		@Override
		public void run() {
			String topic = this.messageQueue.getTopic();

			// 1、检查cancelled变量是为true，若为false则直接退出该线程；否则继续下面的处理；
			if (!this.isCancelled()) {

				// 2、以MessageQueue对象的topic值从MQPullConsumerScheduleService.callbackTable变量中获取PullTaskCallback的实现类（该类是由应用层实现）；
				PullTaskCallback pullTaskCallback = MQPullConsumerScheduleService.this.callbackTable.get(topic);
				if (pullTaskCallback != null) {
					final PullTaskContext context = new PullTaskContext();
					context.setPullConsumer(MQPullConsumerScheduleService.this.defaultMQPullConsumer);
					try {

						// 3、调用该PullTaskCallback实现类的doPullTask方法，
						// 即实现业务层定义的业务逻辑（通用逻辑是先获取消息内容，然后进行相应的业务处理，最后更新消费进度）；
						pullTaskCallback.doPullTask(this.messageQueue, context);
					} catch (Throwable e) {
						context.setPullNextDelayTimeMillis(1000);
						log.error("doPullTask Exception", e);
					}

					// 4、再次检查cancelled变量是为true，
					if (!this.isCancelled()) {
						// 若不为true，则将该PullTaskImpl对象再次放入MQPullConsumerScheduleService. scheduledThreadPoolExecutor线程池中，
						// 设定在200毫秒之后重新调度执行PullTaskImpl线程类；
						MQPullConsumerScheduleService.this.scheduledThreadPoolExecutor.schedule(this, context.getPullNextDelayTimeMillis(), TimeUnit.MILLISECONDS);
					} else {
						log.warn("The Pull Task is cancelled after doPullTask, {}", messageQueue);
					}
				} else {
					log.warn("Pull Task Callback not exist , {}", topic);
				}
			} else {
				log.warn("The Pull Task is cancelled, {}", messageQueue);
			}
		}

		public boolean isCancelled() {
			return cancelled;
		}

		public void setCancelled(boolean cancelled) {
			this.cancelled = cancelled;
		}

		public MessageQueue getMessageQueue() {
			return messageQueue;
		}
	}

	public MQPullConsumerScheduleService(final String consumerGroup) {
		this.defaultMQPullConsumer = new DefaultMQPullConsumer(consumerGroup);
		this.defaultMQPullConsumer.setMessageModel(MessageModel.CLUSTERING);
	}

	public void putTask(String topic, Set<MessageQueue> mqNewSet) {
		// 表示正在拉取消息的任务列表
		Iterator<Entry<MessageQueue, PullTaskImpl>> it = this.taskTable.entrySet().iterator();
		while (it.hasNext()) {
			Entry<MessageQueue, PullTaskImpl> next = it.next();
			if (next.getKey().getTopic().equals(topic)) {
				// 1、检查该topic下面的所有MessageQueue对象
				if (!mqNewSet.contains(next.getKey())) {
					// 若该对象不在入参mqNewSet集合中的，将对应的PullTaskImpl对象的cancelled变量标记为true；
					next.getValue().setCancelled(true);
					it.remove();
				}
			}
		}

		// 2、对于mqNewSet集合中的MessageQueue对象，
		for (MessageQueue mq : mqNewSet) {
			if (!this.taskTable.containsKey(mq)) {
				// 若不在MQPullConsumerScheduleService.taskTable列表中，
				// 则以MessageQueue对象为参数初始化PullTaskImpl对象（该对象为Runnable线程）
				// **************************拉取消息的线程**********************************
				PullTaskImpl command = new PullTaskImpl(mq);

				// 然后放入taskTable列表中
				this.taskTable.put(mq, command);

				// 并将该PullTaskImpl对象放入MQPullConsumerScheduleService.scheduledThreadPoolExecutor线程池中，然后立即执行该线程。
				this.scheduledThreadPoolExecutor.schedule(command, 0, TimeUnit.MILLISECONDS);
			}
		}
	}

	public void start() throws MQClientException {
		final String group = this.defaultMQPullConsumer.getConsumerGroup();
		this.scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(//
				this.pullThreadNums, //
				new ThreadFactoryImpl("PullMsgThread-" + group)//
		);

		this.defaultMQPullConsumer.setMessageQueueListener(this.messageQueueListener);

		this.defaultMQPullConsumer.start();

		log.info("MQPullConsumerScheduleService start OK, {} {}", this.defaultMQPullConsumer.getConsumerGroup(), this.callbackTable);
	}

	public void registerPullTaskCallback(final String topic, final PullTaskCallback callback) {
		this.callbackTable.put(topic, callback);
		this.defaultMQPullConsumer.registerMessageQueueListener(topic, null);
	}

	public void shutdown() {
		if (this.scheduledThreadPoolExecutor != null) {
			this.scheduledThreadPoolExecutor.shutdown();
		}

		if (this.defaultMQPullConsumer != null) {
			this.defaultMQPullConsumer.shutdown();
		}
	}

	public ConcurrentHashMap<String, PullTaskCallback> getCallbackTable() {
		return callbackTable;
	}

	public void setCallbackTable(ConcurrentHashMap<String, PullTaskCallback> callbackTable) {
		this.callbackTable = callbackTable;
	}

	public int getPullThreadNums() {
		return pullThreadNums;
	}

	public void setPullThreadNums(int pullThreadNums) {
		this.pullThreadNums = pullThreadNums;
	}

	public DefaultMQPullConsumer getDefaultMQPullConsumer() {
		return defaultMQPullConsumer;
	}

	public void setDefaultMQPullConsumer(DefaultMQPullConsumer defaultMQPullConsumer) {
		this.defaultMQPullConsumer = defaultMQPullConsumer;
	}

	public MessageModel getMessageModel() {
		return this.defaultMQPullConsumer.getMessageModel();
	}

	public void setMessageModel(MessageModel messageModel) {
		this.defaultMQPullConsumer.setMessageModel(messageModel);
	}
}
